use super::{
    err, err_value, get_backup_root, get_restore_root, get_root, json, ok, throw_err, CSend, ECode,
    RNull, Resp, VBase, VMap, Value, VecS, VecSv,
};
use chrono::prelude::*;
use rocksdb::{
    backup::{BackupEngine, BackupEngineOptions, RestoreOptions},
    DBCompressionType, Env, Options, SliceTransform, WriteBatch, DB,
};
use tracing::{instrument, warn};

pub struct Rdb {
    client: DB,
}

impl Rdb {
    pub fn new() -> Result<Self, Value> {
        let path = get_root();

        let mut opt = Options::default();

        opt.set_level_compaction_dynamic_level_bytes(true);
        opt.set_max_background_jobs(6);
        opt.set_bytes_per_sync(1024 * 1024);
        opt.set_compression_type(DBCompressionType::Lz4);
        opt.set_prefix_extractor(SliceTransform::create_fixed_prefix(3));
        opt.create_if_missing(true);

        let db = DB::open(&opt, path).or_else(|x| err!(ECode::Db, x))?;

        Ok(Rdb { client: db })
    }

    pub fn get(&self, k: &str) -> Resp {
        match self.client.get(k) {
            Ok(Some(value)) => {
                let val = String::from_utf8(value).or_else(|x| err!(ECode::Db, x))?;
                let body = serde_json::from_str(&val).or_else(|x| err!(ECode::Db, x))?;

                Ok(body)
            }
            Ok(None) => err!(ECode::Db, "value not found"),
            Err(e) => err!(ECode::Db, e),
        }
    }

    pub fn batch_que(&self, data: Value) -> RNull {
        let val = data.ckarray()?;

        let mut put = VecSv::new();
        let mut del = VecS::new();

        for d in val.iter() {
            let k = d.ck_str("k")?;
            match d.k_val("v") {
                Some(v) => {
                    put.push((k.to_string(), v.clone()));
                }
                None => {
                    del.push(k.to_string());
                }
            }
        }

        self.batch(Some(&put), Some(&del))
    }

    pub async fn put_que(&self, p: Option<&VecSv>, d: Option<&VecS>, tx: &CSend) -> RNull {
        let mut ret = Vec::new();

        if let Some(data) = p {
            for d in data.iter() {
                ret.push(json!({
                    "k": d.0,
                    "v": d.1,
                }));
            }
        }

        if let Some(data) = d {
            for d in data.iter() {
                ret.push(json!({
                    "k": d,
                }));
            }
        }

        if ret.is_empty() {
            return Ok(());
        }

        tx.send(json!(ret)).await.or_else(|x| err!(ECode::Db, x))?;

        Ok(())
    }

    pub fn batch(&self, p: Option<&VecSv>, d: Option<&VecS>) -> RNull {
        let mut batch = WriteBatch::default();

        if let Some(data) = p {
            for d in data.iter() {
                batch.put(&d.0, d.1.to_string());
            }
        }

        if let Some(data) = d {
            for d in data.iter() {
                batch.delete(&d);
            }
        }

        if batch.is_empty() {
            return Ok(());
        }

        self.client.write(batch).or_else(|e| err!(ECode::Db, e))?;

        Ok(())
    }
}

impl Rdb {
    #[instrument(skip_all, name = "flush")]
    pub fn flush(&self) -> RNull {
        warn!("flush memtables to SST file");
        self.client.flush().or_else(|x| err!(ECode::Db, x))?;

        warn!("flush WAL buffer to disk");
        self.client
            .flush_wal(true)
            .or_else(|x| err!(ECode::Db, x))?;

        Ok(())
    }

    fn open_backup(&self, path: &str) -> Result<BackupEngine, Value> {
        let root = get_backup_root();
        let p = format!("{root}/{path}");

        let opt = BackupEngineOptions::new(&p).or_else(|x| err!(ECode::BackUp, x))?;
        let env = Env::new().or_else(|x| err!(ECode::BackUp, x))?;

        BackupEngine::open(&opt, &env).or_else(|x| err!(ECode::BackUp, x))
    }

    pub fn del_bk(&self, path: &str, num: i64) -> RNull {
        let mut bk = self.open_backup(path)?;

        bk.purge_old_backups(num as usize)
            .or_else(|x| err!(ECode::BackUp, x))?;

        self.flush()
    }

    pub fn restore(&self, path: &str, id: Option<i64>) -> RNull {
        let mut bk = self.open_backup(path)?;

        let mut opt = RestoreOptions::default();
        opt.set_keep_log_files(true);

        let p = get_restore_root();

        match id {
            Some(v) => {
                bk.restore_from_backup(&p, &p, &opt, v as u32)
                    .or_else(|x| err!(ECode::BackUp, x))?;
            }
            None => {
                bk.restore_from_latest_backup(&p, &p, &opt)
                    .or_else(|x| err!(ECode::BackUp, x))?;
            }
        }

        self.flush()
    }

    pub fn backup_info(&self, path: &str) -> Resp {
        let bk = self.open_backup(path)?;

        let tmp = bk.get_backup_info();

        let mut ret = Vec::new();

        for d in tmp {
            let time = match DateTime::from_timestamp_millis(d.timestamp) {
                Some(v) => v,
                None => throw_err!(ECode::BackUp, "timestamp error"),
            };

            let k = d.size / 1024;
            let m = k / 1024;
            let g = m / 1024;

            ret.push(json!({
                "id": d.backup_id,
                "size": format!("{}G, {}M, {}K, {}", g, m, k, d.size),
                "filenum": d.num_files,
                "tm": d.timestamp,
                "time": time.format("%Y-%m-%d %H:%M:%S").to_string(),
            }));
        }

        ok!(ret)
    }

    pub fn backup(&self, path: &str) -> Resp {
        let mut bk = self.open_backup(path)?;

        bk.create_new_backup_flush(&self.client, true)
            .or_else(|x| err!(ECode::BackUp, x))?;

        ok!()
    }
}
